/*
  Copyright 2012 Sergey Ostanin

  Licensed under the Apache License, Version 2.0 (the "License");
  you may not use this file except in compliance with the License.
  You may obtain a copy of the License at

      http://www.apache.org/licenses/LICENSE-2.0

  Unless required by applicable law or agreed to in writing, software
  distributed under the License is distributed on an "AS IS" BASIS,
  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  See the License for the specific language governing permissions and
  limitations under the License.
*/

package ru.irenproject.store;

import java.nio.file.Path;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.commons.lang3.Validate;

import com.almworks.sqlite4java.SQLiteConnection;
import com.almworks.sqlite4java.SQLiteException;
import com.almworks.sqlite4java.SQLiteStatement;
import com.google.common.util.concurrent.Uninterruptibles;

public final class Checkpointer extends Thread {
  private static final int CHECKPOINT_DELAY_SECONDS = 1;

  public static Checkpointer create(Path databaseFile) {
    Checkpointer res = new Checkpointer(databaseFile);
    res.start();
    res.fInitSemaphore.acquireUninterruptibly();
    Validate.isTrue(res.fInitialized);
    return res;
  }

  private final Path fDatabaseFile;
  private volatile boolean fFinish;
  private final AtomicLong fSequenceNumber = new AtomicLong();
  private long fCheckpointedSequenceNumber;

  private final Semaphore fInitSemaphore = new Semaphore(0);
  private volatile boolean fInitialized;

  private Checkpointer(Path databaseFile) {
    super(Checkpointer.class.getName());
    fDatabaseFile = Validate.notNull(databaseFile);
  }

  @Override public void run() {
    try {
      SQLiteConnection conn = new SQLiteConnection(fDatabaseFile.toFile());
      try {
        conn.open(false);
        SQLiteStatement st = conn.prepare("PRAGMA wal_checkpoint");
        try {
          fInitialized = true;
          fInitSemaphore.release();

          while (!fFinish) {
            try {
              sleep(CHECKPOINT_DELAY_SECONDS * 1000);
              long seq = fSequenceNumber.get();
              if (seq != fCheckpointedSequenceNumber) {
                st.step();
                fCheckpointedSequenceNumber = seq;
              }
            } catch (InterruptedException e) {
              // the current iteration was interrupted - just continue to the next one
            } catch (Exception e) {
              e.printStackTrace();
            }
          }
        } finally {
          st.dispose();
        }
      } finally {
        conn.dispose();
      }
    } catch (SQLiteException e) {
      throw new RuntimeException(e);
    } finally {
      fInitSemaphore.release();
    }
  }

  public void finish() {
    fFinish = true;
    interrupt();
    Uninterruptibles.joinUninterruptibly(this);
  }

  public void advanceSequenceNumber() {
    fSequenceNumber.incrementAndGet();
  }

  public void restartCountdown() {
    interrupt();
  }
}
